package core;

import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.net.DatagramPacket;
import java.net.InetAddress;
import java.net.MulticastSocket;
import java.net.Socket;
import java.net.UnknownHostException;
import java.util.Collections;
import java.util.Date;
import java.util.Iterator;
import java.util.Vector;

import lib.Logger;
import lib.LostMessage;
import lib.Message;
import lib.Payload;
import lib.RequestMessage;
import lib.Symbols;
import lib.VectorClock;

public class CausalHandler extends Thread {

	private MulticastSocket multicastSocket;
	private InetAddress group;
	private String groupIp;
	private int port;
	private VectorClock vectorClock;
	private String ip;
	private String processId;

	private MessageListener messageListener;
	private Vector<LostMessage> lostMessages;
	private Vector<LostMessage> recoveredMessages;
	private Vector<Message> queuedMessages;
	private boolean logoff;
	
	private RepeaterHandler repeaterHandler;

	public CausalHandler(MessageListener messageListener, String groupIp, int port, int ttl) throws IOException {
		this.groupIp = groupIp;
		this.port = port;
		this.group = InetAddress.getByName(groupIp);
		this.multicastSocket = new MulticastSocket(port);
		this.multicastSocket.setTimeToLive(ttl);
		this.vectorClock = new VectorClock();
		this.ip = InetAddress.getLocalHost().getHostAddress();
		
		this.messageListener = messageListener;
		this.lostMessages = new Vector<LostMessage>();
		this.recoveredMessages = new Vector<LostMessage>();
		this.queuedMessages = new Vector<Message>();
		this.logoff = false;
		
		this.repeaterHandler = new RepeaterHandler();
	}

	public void sendMessage(Payload payload) throws IOException {
		vectorClock.addOneTo(processId);
		int messageId = vectorClock.get(processId);
		VectorClock vectorClockClone = (VectorClock) vectorClock.Clone();
		Message message = new Message(ip, processId, messageId, vectorClockClone, payload);
		
		byte[] buffer = MessageHandler.getByteFrom(message);
		DatagramPacket datagramPacket = new DatagramPacket(buffer, buffer.length, group, port);
		
		multicastSocket.send(datagramPacket);
		Logger.println("Broadcasted message " + message.toString());
		
		repeaterHandler.addToHistory(message);
	}
	
	public Payload receiveMessage(int length) throws IOException {
		byte[] buffer = new byte[length];
		DatagramPacket datagramPacket = new DatagramPacket(buffer, buffer.length);
		
		multicastSocket.receive(datagramPacket);
		Message message = MessageHandler.getMessageFrom(datagramPacket.getData());
		message.setReceiveDate(new Date());
		
		if(Math.random()<Symbols.MISSING_PROB && !message.getProcessId().equals(processId)) {
			Logger.println("Missed message " + message.toString() + " - " + datagramPacket.getLength() + " bytes");
		} else {
			Logger.println("Received message " + message.toString() + " - " + datagramPacket.getLength() + " bytes");

			if(!isDuplicate(message)) {
				while(isCorrupted(message)) {
					if(message.getProcessId().equals(processId)) {
						message = repeaterHandler.findMessageInHistory(message.getDigest());
					} else {
						message = requestMessage(message.getIp(), message.getDigest(), length);
					}
				}
				if(isEligible(message, true)) {
					return message.getPayload();
				} else {
					addInQueue(message);
				}
			}
		}
		
		return null;
	}
	
	private synchronized boolean isDuplicate(Message message) {
		for(LostMessage m : recoveredMessages) {
			if(m.getProcessId().equals(message.getProcessId()) && m.getMessageId()==message.getMessageId()) {
				return true;
			}
		}
		return false;
	}
	
	private boolean isCorrupted(Message message) throws IOException {
		byte[] receivedDigest = message.getDigest();
		byte[] computedDigest = IntegrityHandler.getDigest(message);
		
		boolean different = !(IntegrityHandler.compareDigests(receivedDigest, computedDigest));
		boolean corrupted = (different || Math.random()<Symbols.CORRUPTED_PROB);
		Logger.println("Checking integrity of message " + message.toString()  + " - corrupted: " + corrupted + " - received " + IntegrityHandler.byteToHexString(receivedDigest) + " computed " + IntegrityHandler.byteToHexString(computedDigest));
		return corrupted;
	}
	
	private synchronized void addInQueue(Message message) {
		if(!queuedMessages.contains(message)) {
			queuedMessages.add(message);
			Collections.sort(queuedMessages);
		}
	}
	
	private boolean isEligible(Message message, boolean request) throws IOException {
		boolean eligible = true;
		
		if(!message.getProcessId().equals(processId)) {
			VectorClock piggybackedVectorClock = message.getPiggybackedVectorClock();
			Iterator<String> it = piggybackedVectorClock.getProcessIds().iterator();
			int seen = vectorClock.get(message.getProcessId());
			int sent = piggybackedVectorClock.get(message.getProcessId());
			eligible = (seen == sent-1);
			
			if(!eligible && request) {
				Logger.println("Lost messages from " + (seen+1) + " to " + (sent-1) + " of " + message.getProcessId());
				addLostMessages(message.getProcessId(), message.getIp(), seen+1, sent-1);
			}
			
			String process;
			while(it.hasNext() && eligible) {
				process = it.next();
				if(!process.equals(message.getProcessId())) {
					eligible = (vectorClock.get(process) >= piggybackedVectorClock.get(process));
				}
			}
			if(eligible) {
				vectorClock = VectorClock.mergeClocks(piggybackedVectorClock, vectorClock);
			}
		}
		Logger.println("Checking causality of message " + message.toString() + " - eligible: " + eligible);
		return eligible;
	}
	
	private synchronized void addLostMessages(String processId, String ip, int fromId, int toId) {
		for(int i=fromId; i<=toId; i++) {
			lostMessages.add(new LostMessage(processId, ip, i, new Date()));
		}
	}
	
	private synchronized Vector<Message> getEligibleQueuedMessage() throws IOException {
		Vector<Message> messages = new Vector<Message>();
		
		if(queuedMessages.size()>0) {
			Logger.println("There are " + queuedMessages.size() + " queued messages");
		}
		for(Message m : queuedMessages) {
			if(isEligible(m, false)) {
				messages.add(m);
			}
		}
		queuedMessages.removeAll(messages);
		return messages;
	}
	
	private Message requestMessage(String ip, byte[] digest, int length) throws IOException {
		RequestMessage request = new RequestMessage(digest);
		return doSocketCall(ip, request, length);
	}
	
	private Message requestMessage(String ip, int messageId, int length) throws IOException {
		RequestMessage request = new RequestMessage(messageId);
		return doSocketCall(ip, request, length);
	}
	
	private Message doSocketCall(String ip, RequestMessage request, int length) throws IOException {
		byte[] buffer = new byte[length];
		Message message;
		Socket socket = new Socket(ip, Symbols.REPEATER_PORT+Symbols.OFFSET_2);
		InputStream in = socket.getInputStream();
		OutputStream out = socket.getOutputStream();
		
		out.write(MessageHandler.getByteFrom(request));
		out.flush();
		in.read(buffer);
		message = MessageHandler.getMessageFrom(buffer);
		socket.close();
		in.close();
		out.close();
		
		return message;
	}
	
	public void logon(String nickName) throws IOException {
		processId = nickName + "/" + ip;
		Logger.println("Joined group " + groupIp + ":" + port + " with id " + processId);
		multicastSocket.joinGroup(group);
		repeaterHandler.start();
		this.start();
	}
	
	public void logoff() throws IOException {
		Logger.println("Left group " + groupIp + ":" + port);
		multicastSocket.leaveGroup(group);
		multicastSocket.close();
		repeaterHandler.logoff();
		this.logoff = true;
	}
	
	public void run() {
		Message message;
		Vector<Message> messages;
		Vector<Payload> payloads = new Vector<Payload>();
		String log = "";
		
		while(!logoff) {
			for(LostMessage lost : lostMessages) {
				if(new Date().getTime()-lost.getDate().getTime()>Symbols.MAX_WAIT*1000) {
					log = "Pulling message " + lost.toString() + " " + "[" + (lostMessages.indexOf(lost)+1) + "/" + lostMessages.size() + "]";
					try {
						message = requestMessage(lost.getIp(), lost.getMessageId(), Symbols.MAX_BUFFER_DIM);
						if(message==null) {
							log += " - error - message null";
						} else {
							message.setReceiveDate(new Date());
							log += " - success";
							addInQueue(message);
							recoveredMessages.add(lost);
						}
					} catch (UnknownHostException e) {
						log += " - error - UnknownHostException - " + e.getMessage();
					} catch (IOException e) {
						log += " - error - IOException - " + e.getMessage();
					}
					Logger.println(log);
				}
			}
			if(lostMessages.size()>0) {
				lostMessages.removeAll(recoveredMessages);
				Logger.println("Recovered " + recoveredMessages.size() + " messages - " + lostMessages.size() + " remaining");
			}
			try {
				messages = getEligibleQueuedMessage();
				if(messages.size()>0) {
					for(Message m : messages) {
						payloads.add(m.getPayload());
					}
					Logger.println("Pushing " + messages.size() + " messages back to application");
					messageListener.pushMessages(payloads);
				}
			} catch (IOException e) {
				//e.printStackTrace();
				Logger.println("IOException in Causalhandler Thread: " + e.getMessage());
			}
			try {
				Thread.sleep(Symbols.SLEEP*1000);
			} catch (InterruptedException e) {
				//e.printStackTrace();
				Logger.println("InterruptedException in Causalhandler Thread: " + e.getMessage());
			}
		}
	}

}
